package com.gjy.amqp.simple.delayed;

import com.gjy.amqp.util.RabbitmqUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-05-17 11:18:56
 */
public class DelayedProducer {
    private static final Logger log = LoggerFactory.getLogger(DelayedProducer.class);
    private static final String DELAYED_EXCHANGE = "delayed_exchange";
    private static final String DELAYED_QUEUE = "delayed_queue";
    private static final String DELAYED_DLX_EXCHANGE = "dlx_exchange";
    private static final String DELAYED_DLX_QUEUE = "dlx_queue";

    public static void main(String[] args) throws Exception {
        Connection connection = RabbitmqUtil.connection();
        Channel channel = connection.createChannel();

        // 创建普通交换机
        channel.exchangeDeclare(DELAYED_EXCHANGE, BuiltinExchangeType.DIRECT, true);
        HashMap<String, Object> queueArgs = new HashMap<>();
        // 队列内ttl最大10秒，等待超时后送入x-dead-letter-exchange死信交换机
        queueArgs.put("x-message-ttl", 10000);
        queueArgs.put("x-dead-letter-exchange", DELAYED_DLX_EXCHANGE);
        // 创建普通队列
        channel.queueDeclare(DELAYED_QUEUE, true, false, false, queueArgs);
        // 普通队列和普通交换机绑定
        channel.queueBind(DELAYED_QUEUE, DELAYED_EXCHANGE, "");

        // 创建死信交换机
        channel.exchangeDeclare(DELAYED_DLX_EXCHANGE, BuiltinExchangeType.DIRECT, true);
        // 创建死信队列
        channel.queueDeclare(DELAYED_DLX_QUEUE, true, false, false, new HashMap<>());
        // 绑定私信队列
        channel.queueBind(DELAYED_DLX_QUEUE, DELAYED_DLX_EXCHANGE, "");

        for (int i = 0; i < 10; i++) {
            String s = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd HH:mm:ss"));
            channel.basicPublish(DELAYED_EXCHANGE, "", null, s.getBytes(StandardCharsets.UTF_8));
            log.info("消息: {} 已发送", s);
            TimeUnit.SECONDS.sleep(1);
        }

        channel.close();
        connection.close();
    }

}
